package org.example.flinksql.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 单表过滤、拷贝demo
 * // total 60000 records
 * // start time: 10:53:24,964
 * // end time: 10:53:54,779
 */
public class FilterAndCopySingleTable {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // CREATE TEMPORARY TABLE t_users
        tbEnv.executeSql("CREATE TABLE t_users (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // CREATE TEMPORARY TABLE t_users_man
        // TODO:Sink必须设置主键，当主键冲突时会自动更新数据，如果不设置程序会报错
        tbEnv.executeSql("CREATE TABLE t_users_man (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT," +
                "PRIMARY KEY (id) NOT ENFORCED) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users_man'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // CREATE TEMPORARY TABLE t_users_women
        // TODO:Sink必须设置主键，当主键冲突时会自动更新数据，如果不设置程序会报错
        tbEnv.executeSql("CREATE TABLE t_users_women (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT," +
                "PRIMARY KEY (id) NOT ENFORCED) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users_women'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");


        // 直接使用INSERT INTO ... SELECT
//        tbEnv.executeSql("INSERT INTO t_users_man select * from t_users where sex = '男'");
//        tbEnv.executeSql("INSERT INTO t_users_women select * from t_users where sex = '女'");

        // 使用table api: executeInsert
        Table table = tbEnv.sqlQuery("select * from t_users where sex = '男'");
        Table table1 = tbEnv.sqlQuery("select * from t_users where sex = '女'");
        table.executeInsert("t_users_man").print();
        table1.executeInsert("t_users_women").print();

        // TODO:delete not support
//        tbEnv.executeSql("delete from t_users_man where id = 1");

        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
